/**
 *
 * Copyright 2009 garrett
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use
 * this file except in compliance with the License.  You may obtain a copy of the
 * License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software distributed
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied. See the License for the specific
 * language governing permissions and limitations under the License. 
 */
package org.memesis.pandorasbook.util;

import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * Utility methods for dealing with ActiveMQ.
 *
 * @author garrett
 */
public class ActiveMQHelper {
	private static Map<String, Connection> connectionMap = new HashMap<String, Connection>();
	
	/**
	 * Given a URL for some ActiveMQ broker, return an existing connection or create a new one
	 * 
	 * @param mqUrl the URL for a broker
	 * @return a connection for that broker
	 * @throws JMSException
	 */
	public static Connection getConnection(String mqUrl) throws JMSException {
		Connection conn;
		if(connectionMap.containsKey(mqUrl)) {
			conn = connectionMap.get(mqUrl);
		} else {
			conn = createConnection(mqUrl);
			connectionMap.put(mqUrl, conn);
		}
		return conn;
	}
	
	private static Connection createConnection(String mqUrl) throws JMSException {
		// We assume anonymous login
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, mqUrl);
		Connection conn = factory.createConnection();
		//conn.setExceptionListener(this);
		conn.start();
		return conn;
	}
	
	public static Session createSession(Connection conn) throws JMSException {
		// Assume transactions are always used
		Session sess = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
		return sess;
	}
	
	public static Destination createDestination(Session sess, String subject, boolean isTopic) throws JMSException {
		Destination dest;
		if(isTopic) {
			dest = sess.createTopic(subject);
		} else {
			dest = sess.createQueue(subject);
		}
		return dest;
	}
	
	public static MessageProducer createMessageProducer(Session sess, Destination dest) throws JMSException {
		MessageProducer producer = sess.createProducer(dest);
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		return producer;
	}
	
	public static MessageConsumer createMessageConsumer(Session sess, Destination dest) throws JMSException {
		MessageConsumer consumer = sess.createConsumer(dest);
		// consumer.setMessageListener(....);
		return consumer;
	}
	
	public static void SendTextMessage(Session sess, MessageProducer producer, String message) throws JMSException {
		TextMessage text = sess.createTextMessage(message);
		producer.send(text);
		sess.commit();
	}
}